package net.bwie.realtime.dy.dwd.log.funcation.zl.messagealterfuncation;

import net.bwie.realtime.dy.dwd.log.bean.AlertMessageUser;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：MessageAlertProcessFunction
 * @Author: ZddddQ
 * @Date: 2025/5/28 14:29
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class MessageAlertProcessFunction
        extends ProcessWindowFunction<Integer, AlertMessageUser, String, TimeWindow> {

    private static final int THRESHOLD = 0;
    private final FastDateFormat dateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void process(
            String userId,
            Context context,
            Iterable<Integer> counts,
            Collector<AlertMessageUser> out
    ) {
        // 获取聚合结果
        Integer count = counts.iterator().next();

        // 判断是否超过阈值
        if (count > THRESHOLD) {
            // 获取窗口时间并格式化
            TimeWindow window = context.window();
            String windowStart = dateFormat.format(window.getStart());
            String windowEnd = dateFormat.format(window.getEnd());

            // 构建警报对象
            AlertMessageUser alert = new AlertMessageUser();
            alert.setUserId(userId);
            alert.setWindowStartTime(windowStart);
            alert.setWindowEndTime(windowEnd);
            alert.setMessageCount(count);
            alert.setTs(System.currentTimeMillis());

            out.collect(alert);
        }
    }
}
